#!/usr/bin/env python3

# Copyright (c) 2005-2014 Fpemud <fpemud@sina.com>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.


import os
import re
import io
import sys
import uuid
import glob
import gzip
import time
import lxml.html
import mrget
import struct
import shutil
import psutil
import pygit2
import urllib.request
import pathlib
import hashlib
import filecmp
import subprocess


class Util:

    @staticmethod
    def doNothing(*kargs, **kwargs):
        pass

    @staticmethod
    def realPathSplit(path):
        """os.path.split() only split a path into 2 component, I believe there are reasons, but it is really inconvenient.
           So I write this function to split a unix path into basic components.
           Reference: http://stackoverflow.com/questions/3167154/how-to-split-a-dos-path-into-its-components-in-python"""

        folders = []
        while True:
            path, folder = os.path.split(path)
            if folder != "":
                folders.append(folder)
            else:
                if path != "":
                    folders.append(path)
                break
        if path.startswith("/"):
            folders.append("")
        folders.reverse()
        return folders

    @staticmethod
    def fullyNormPath(path):
        # this is what we need:
        #   os.path.normpath(os.path.join(os.path.normpath(path), "foo")) == os.path.join(os.path.normpath(path), "foo")
        # these are what we have:
        #   os.path.join(".", "foo") -> "./foo"
        #   os.path.join("", "foo") -> "foo"
        #   os.path.normpath("./foo") -> "foo"
        #   os.path.normpath("foo") -> "foo"
        # we have to write this function, to make:
        #   os.path.normpath(os.path.join(Util.fullyNormPath(path), "foo")) == os.path.join(Util.fullyNormPath(path), "foo")
        path = os.path.normpath(path)
        return "" if path == "." else path

    @staticmethod
    def extractDeb(fullfn, compressType):
        if compressType == "xz":
            subprocess.check_call("ar p '%s' data.tar.xz | tar -xJ" % (fullfn), shell=True)
        else:
            assert False

    @staticmethod
    def fetchAndParseHtmlPage(myName, url):
        if url.startswith("mirror://"):
            url = mrget.target_urls(url)[0]

        while True:
            try:
                with urllib.request.urlopen(url, timeout=60) as resp:
                    if resp.info().get('Content-Encoding') is None:
                        fakef = resp
                    elif resp.info().get('Content-Encoding') == 'gzip':
                        fakef = io.BytesIO(resp.read())
                        fakef = gzip.GzipFile(fileobj=fakef)
                    else:
                        assert False
                    return lxml.html.parse(fakef)
            except OSError as e:
                print("%s: Failed to acces %s, %s" % (myName, url, e))
                time.sleep(10)

    @staticmethod
    def fetchHtmlBinaryData(myName, url):
        if url.startswith("mirror://"):
            url = mrget.target_urls(url)[0]

        while True:
            try:
                with urllib.request.urlopen(url, timeout=60) as resp:
                    return resp.read()
            except OSError as e:
                print("%s: Failed to acces %s, %s" % (myName, url, e))
                time.sleep(10)

    @staticmethod
    def sed(fn, pattern, repl, regex=False, exceptionClass=None):
        buf = pathlib.Path(fn).read_text()

        if not regex:
            if pattern not in buf and exceptionClass is not None:
                raise exceptionClass("Can not find the specified pattern")
            buf = buf.replace(pattern, repl)
        else:
            buf = re.sub(pattern, repl, buf)

        with open(fn, "w") as f:
            f.write(buf)

    @staticmethod
    def iterateLists(*kargs):
        for lst in kargs[1:]:
            assert len(lst) == len(kargs[0])
        for i in range(0, len(kargs[0])):
            yield tuple([lst[i] for lst in kargs])

    @staticmethod
    def mntOptsHasPattern(opts, pattern):
        tlist = opts.split(",")
        return any([re.fullmatch(pattern, x) is not None for x in tlist])

    @staticmethod
    def mntOptsRemovePattern(opts, pattern):
        tlist = opts.split(",")
        tlist = [x for x in tlist if re.fullmatch(pattern, x) is None]
        return ",".join(tlist)

    @staticmethod
    def kmodGetModuleInfoByFile(path):
        return Util._kmodGetModuleInfo(None, path)

    @staticmethod
    def kmodGetModuleInfoByAlias(kernelModuleDir, alias):
        assert kernelModuleDir.startswith("/lib/modules/")
        return Util._kmodGetModuleInfo(os.path.basename(kernelModuleDir), alias)

    @staticmethod
    def _kmodGetModuleInfo(k, fileOrAlias):
        ret = {
            "filepath": None,
            "firmwares": [],
            "depends": [],
            "softdep-pre": [],
        }

        if k is not None:
            opt = ["-k", k]
        else:
            opt = []

        for line in subprocess.check_output(["modinfo", *opt, fileOrAlias], text=True).split("\n"):
            if True:
                m = re.fullmatch("filename: +(\\S.*)", line)
                if m is not None:
                    assert ret["filepath"] is None
                    ret["filepath"] = m.group(1)
                    continue
            if True:
                m = re.fullmatch("firmware: +(\\S.*)", line)
                if m is not None:
                    ret["firmwares"].append(m.group(1))
                    continue
            if True:
                m = re.fullmatch("depends: +(\\S.*)", line)
                if m is not None:
                    assert ret['depends'] == []
                    ret["depends"] = m.group(1).split(",")
                    continue
            if True:
                m = re.fullmatch("softdep: +pre: (\\S.*)", line)
                if m is not None:
                    ret["softdep-pre"].append(m.group(1))
                    continue
            if True:
                # ignore softdep-post
                pass
            if True:
                m = re.fullmatch("softdep: +preunload: (\\S.*)", line)
                if m is not None:
                    raise Exception("softdep preunload in %s is not supported" % (fileOrAlias))
            if True:
                m = re.fullmatch("softdep: +postunload: (\\S.*)", line)
                if m is not None:
                    raise Exception("softdep postunload in %s is not supported" % (fileOrAlias))

        return ret

    @staticmethod
    def kmodGetKmodAndDeps(kernelModuleDir, kmodAlias, withDeps):
        kmodList = dict()                                                               # use dict to remove duplication while keeping order
        Util._kmodGetKmodAndDepsImpl(kernelModuleDir, kmodAlias, withDeps, kmodList)
        return list(kmodList.keys())

    @staticmethod
    def _kmodGetKmodAndDepsImpl(kernelModuleDir, kmodAlias, withDeps, result):
        ret = Util.kmodGetModuleInfoByAlias(kernelModuleDir, kmodAlias)

        if withDeps:
            for kmodAlias in ret["depends"]:
                Util._kmodGetKmodAndDepsImpl(kernelModuleDir, kmodAlias, withDeps, result)
            for kmodAlias in ret["softdep-pre"]:
                Util._kmodGetKmodAndDepsImpl(kernelModuleDir, kmodAlias, withDeps, result)

        if ret["filepath"] is not None and ret["filepath"] != "(builtin)":
            # this module is not built into the kernel
            result[ret["filepath"]] = None

    @staticmethod
    def libUsed(binFile):
        """Return a list of the paths of the shared libraries used by binFile"""

        LDD_STYLE1 = re.compile(r'^\t(.+?)\s\=\>\s(.+?)?\s\(0x.+?\)$')
        LDD_STYLE2 = re.compile(r'^\t(.+?)\s\(0x.+?\)$')

        try:
            raw_output = subprocess.check_output(["/usr/bin/ldd", "--", binFile], stderr=subprocess.STDOUT, text=True)
        except subprocess.CalledProcessError as e:
            if 'not a dynamic executable' in e.output:
                raise Exception("not a dynamic executable")
            else:
                raise

        # We can expect output like this:
        # [tab]path1[space][paren]0xaddr[paren]
        # or
        # [tab]path1[space+]=>[space+]path2?[paren]0xaddr[paren]
        # path1 can be ignored if => appears
        # path2 could be empty

        if 'statically linked' in raw_output:
            return []

        result = []
        for line in raw_output.splitlines():
            match = LDD_STYLE1.match(line)
            if match is not None:
                if match.group(2):
                    result.append(match.group(2))
                continue

            match = LDD_STYLE2.match(line)
            if match is not None:
                result.append(match.group(1))
                continue

            assert False

        result.remove("linux-vdso.so.1")
        return result

    @staticmethod
    def verifyChecksum(file_path, algorithm, checksum):
        if algorithm == "sha256":
            return Util.verifySha256(file_path, checksum)
        else:
            assert False

    @staticmethod
    def verifySha256(file_path, expected_checksum):
        sha256_hash = hashlib.sha256()
        with open(file_path, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                sha256_hash.update(chunk)
        return sha256_hash.hexdigest() == expected_checksum

    @staticmethod
    def gitHasCommitOrTag(path, commit_or_tag):
        repo = pygit2.Repository(path)
        if commit_or_tag.startswith("refs/tags/"):        # tag
            try:
                repo.revparse_single(commit_or_tag)
                return True
            except KeyError:
                return False
        else:                                             # commit hash
            try:
                repo[commit_or_tag]
                return True
            except KeyError:
                return False

    @staticmethod
    def gitConfigGet(dirName, name):
        # FIXME: should change to use libgit2
        gitDir = os.path.join(dirName, ".git")
        cmdStr = "GIT_PAGER=cat git --git-dir=\"%s\" --work-tree=\"%s\" config --get-all %s" % (gitDir, dirName, name)
        ret = subprocess.check_output(cmdStr, shell=True, text=True).rstrip("\n")

        if name in ["remote.origin.fetch"]:
            return ret.split("\n")

        assert "\n" not in ret
        return ret

    @classmethod
    def gitConfigCompare(cls, dirName, name, value):
        if name == "remote.origin.fetch":
            if not isinstance(value, list):
                value = [value]

            tlist = cls.gitConfigGet(dirName, name)
            for item in value:
                idx = tlist.index(item)
                if idx < 0:
                    return False
                tlist.pop(idx)

            if len(tlist) == 0:
                return True
            elif len(tlist) == 1:
                # it is the default value git auto adds
                # yes it is for repositories that has only one remote, and we only use this kind of repositories
                return tlist[0] == "+refs/heads/*:refs/remotes/origin/*"
            else:
                return False

        if name == "remote.origin.url":
            if not isinstance(value, list):
                value = [value]
            return cls.gitConfigGet(dirName, name) in value

        return cls.gitConfigGet(dirName, name) == value

    @classmethod
    def gitClean(cls, dest_directory):
        for fullfn in glob.glob(os.path.join(dest_directory, ".git", "**", "*.lock"), recursive=True):
            # FIXME:
            # should detect if this lock file is currently occupied by a git process?
            # git simply creates a plain file, or creates a real lock file?
            cls.forceDelete(fullfn)
        subprocess.check_call(["/usr/bin/git", "-C", dest_directory, "reset", "--hard"])  # revert any modifications
        subprocess.check_call(["/usr/bin/git", "-C", dest_directory, "clean", "-xfd"])    # delete untracked files

    @classmethod
    def gitPull(cls, dest_directory, reclone_on_failure=False, clone_url=None, clone_config=None, quiet=False):
        recoverableError = {
            "fatal: unknown index entry format": "match",
            "fatal: refusing to merge unrelated histories": "match",
            "fatal: Updating an unborn branch with changes added to the index.": "match",
            "error: .* did not send all necessary objects": "regex-match",
            "fatal: .* is corrupt": "regex-match",
        }

        if reclone_on_failure:
            if clone_url.startswith("mirror://"):
                mrList = mrget.target_urls(clone_url)
                clone_url = mrList[0]
            else:
                mrList = [clone_url]
        else:
            assert clone_url is None

        if quiet:
            quietArg = "-q"
        elif sys.stderr.isatty():
            quietArg = "--progress"    # Util.shellExec() use pipe to do advanced process, we add "--progress" so that progress can still be displayed
        else:
            quietArg = ""

        mode = "pull"
        while reclone_on_failure:
            if not os.path.exists(dest_directory):
                mode = "clone"
                break
            if not os.path.isdir(os.path.join(dest_directory, ".git")):
                mode = "clone"
                break
            if not cls.gitConfigCompare(dest_directory, "remote.origin.url", mrList):
                mode = "clone"
                break
            if clone_config is not None:
                for k, v in clone_config.items():
                    if not cls.gitConfigCompare(dest_directory, k, v):
                        mode = "clone"
                        break
            break

        while True:
            if mode == "pull":
                try:
                    cls.gitClean(dest_directory)
                except BaseException:
                    if not reclone_on_failure:
                        raise
                    mode = "clone"
                    continue

                try:
                    cmdList = [
                        "git",
                        "-C",
                        dest_directory,
                        "pull",
                        "--rebase",
                        "--no-stat",
                    ]
                    if quietArg != "":
                        cmdList.append(quietArg)
                    subprocess.check_call(cmdList)
                    break
                except subprocess.CalledProcessError as e:
                    bFound = False
                    for recoverableErrorStr, ret in recoverableError.items():
                        if ret == "match":
                            if recoverableErrorStr in str(e.stdout):
                                bFound = True
                                break
                        elif ret == "regex-match":
                            if re.search(recoverableErrorStr, str(e.stdout), re.M):
                                if not reclone_on_failure:
                                    bFound = True
                                    break
                        else:
                            assert False
                    if bFound:
                        if not reclone_on_failure:
                            raise
                        mode = "clone"
                    continue

            if mode == "clone":
                cls.forceDelete(dest_directory)
                if clone_config is not None:
                    configArg = "--config %s" % (" ".join(['%s="%s"' % (k, v) for k, v in clone_config.items()]))       # FIXME: is it space separated?
                else:
                    configArg = ""
                cmd = "git clone %s %s \"%s\" \"%s\"" % (quietArg, configArg, clone_url, dest_directory)
                subprocess.check_call(cmd, shell=True)
                mode = "pull"                            # sometimes we may clone a bad repository which is not pullable, we pull after clone to make sure it is pullable
                continue

            assert False

    @staticmethod
    def truncateDir(path):
        assert os.path.isdir(path)
        for fn in os.listdir(path):
            Util.forceDelete(os.path.join(path, fn))

    @staticmethod
    def forceDelete(path):
        if os.path.islink(path):
            os.remove(path)
        elif os.path.isfile(path):
            os.remove(path)
        elif os.path.isdir(path):
            shutil.rmtree(path)
        elif os.path.lexists(path):
            os.remove(path)             # other type of file, such as device node
        else:
            pass                        # path does not exist, do nothing

    @staticmethod
    def forceMkdir(path):
        if os.path.lexists(path):
            if os.path.isdir(path):
                return
            Util.forceDelete(path)
        os.mkdir(path)

    @staticmethod
    def forceMove(src, dst):
        if os.path.isdir(dst):          # shutil.move() won't overwrite directory, so we delete it first
            if os.path.islink(dst):
                os.remove(dst)
            else:
                shutil.rmtree(dst)
        shutil.move(src, dst)

    @staticmethod
    def forceMoveToDir(src, dstDir):
        assert os.path.isdir(dstDir)
        dst = os.path.join(dstDir, os.path.basename(src))
        Util.forceMove(src, dst)

    def findInList(lst, key):
        for x in lst:
            if key(x):
                return x
        return None

    @staticmethod
    def checkListUnique(lst, key=None):
        if key is None:
            keyList = lst
        else:
            keyList = [key(x) for x in lst]
        return len(keyList) == len(set(keyList))

    @staticmethod
    def globDirRecursively(dirpath, excludeSelf=False):
        # glob.glob("/a/**", recursive=True) returns ["/a/", "/a/a", "/a/a/a", ...]
        # the first element sucks, normalize it
        ret = glob.glob(os.path.join(dirpath, "**"), recursive=True)
        assert ret[0] == dirpath + "/"
        if excludeSelf:
            ret.pop(0)
        else:
            ret[0] = dirpath
        return ret

    @staticmethod
    def getBlkDevUuid(devPath):
        """UUID is also called FS-UUID, only device containing a file system has it, note that all btrfs device has same UUID"""

        ret = Util.cmdCall("/sbin/blkid", devPath)
        m = re.search(r'\bUUID="(\S*)"\B', ret, re.M)
        if m is not None:
            return m.group(1)
        return None

    @staticmethod
    def getBlkDevByUuid(uuid):
        path = os.path.join("/dev", "disk", "by-uuid", uuid)
        if not os.path.exists(path):
            return None
        return os.path.realpath(path)

    @staticmethod
    def getBlkDevSubUuid(devPath):
        """UUID_SUB is special UUID for btrfs device"""

        ret = Util.cmdCall("/sbin/blkid", devPath)
        m = re.search(r'\bUUID_SUB="(\S*)"\B', ret, re.M)
        if m is not None:
            return m.group(1)
        return None

    @staticmethod
    def getBlkDevBySubUuid(uuid):
        devName = None
        for line in Util.cmdCall("blkid", "-o", "export").split("\n"):
            m = re.fullmatch("DEVNAME=(\\S+)", line)
            if m is not None:
                devName = m.group(1)
                continue
            m = re.fullmatch("UUID_SUB=(\\S+)", line)
            if m is not None and m.group(1) == uuid:
                break
        return devName

    @staticmethod
    def getBlkDevPartUuid(devPath):
        """only disk partition has PARTUUID"""

        ret = Util.cmdCall("/sbin/blkid", devPath)
        m = re.search(r'\bPARTUUID="(\S*)"\B', ret, re.M)
        if m is not None:
            return m.group(1)
        return None

    @staticmethod
    def getBlkDevByPartUuid(uuid):
        path = os.path.join("/dev", "disk", "by-partuuid", uuid)
        if not os.path.exists(path):
            return None
        return os.path.realpath(path)

    @staticmethod
    def readListFile(filename):
        ret = []
        with open(filename, "r") as f:
            for line in f.read().split("\n"):
                line = line.strip()
                if line != "" and not line.startswith("#"):
                    ret.append(line)
        return ret

    @staticmethod
    def cmdCall(cmd, *kargs):
        # call command to execute backstage job
        #
        # scenario 1, process group receives SIGTERM, SIGINT and SIGHUP:
        #   * callee must auto-terminate, and cause no side-effect
        #   * caller must be terminated by signal, not by detecting child-process failure
        # scenario 2, caller receives SIGTERM, SIGINT, SIGHUP:
        #   * caller is terminated by signal, and NOT notify callee
        #   * callee must auto-terminate, and cause no side-effect, after caller is terminated
        # scenario 3, callee receives SIGTERM, SIGINT, SIGHUP:
        #   * caller detects child-process failure and do appopriate treatment

        ret = subprocess.run([cmd] + list(kargs),
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                             text=True)
        if ret.returncode > 128:
            # for scenario 1, caller's signal handler has the oppotunity to get executed during sleep
            time.sleep(1.0)
        if ret.returncode != 0:
            print(ret.stdout)
            ret.check_returncode()
        return ret.stdout.rstrip()

    @staticmethod
    def cmdCallTestSuccess(cmd, *kargs):
        ret = subprocess.run([cmd] + list(kargs),
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                             text=True)
        if ret.returncode > 128:
            time.sleep(1.0)
        return (ret.returncode == 0)

    @staticmethod
    def bcachefsGetUuid(slaveDevPathList):
        ret = None
        for devPath in slaveDevPathList:
            out = Util.cmdCall("file", "-sb", devPath)
            m = re.search("^bcachefs, UUID=(\\S+),", out)
            if ret is None:
                ret = m.group(1)
            else:
                assert ret == m.group(1)
        return ret

    @staticmethod
    def btrfsGetUuid(mountPoint):
        out = Util.cmdCall("btrfs", "filesystem", "show", mountPoint)
        m = re.search("uuid: (\\S+)", out, re.M)
        return m.group(1)

    @staticmethod
    def btrfsGetSlavePathList(mountPoint):
        ret = []
        out = Util.cmdCall("btrfs", "filesystem", "show", mountPoint)
        for m in re.finditer("path (\\S+)", out, re.M):
            ret.append(m.group(1))
        return ret

    @staticmethod
    def bcacheGetSlaveDevPathList(bcacheDevPath):
        """Last element in the returned list is the backing device, others are cache device"""

        retList = []

        slavePath = "/sys/block/" + os.path.basename(bcacheDevPath) + "/slaves"
        for slaveDev in os.listdir(slavePath):
            retList.append(os.path.join("/dev", slaveDev))

        bcachePath = os.path.realpath("/sys/block/" + os.path.basename(bcacheDevPath) + "/bcache")
        backingDev = os.path.basename(os.path.dirname(bcachePath))
        backingDevPath = os.path.join("/dev", backingDev)

        retList.remove(backingDevPath)
        retList.append(backingDevPath)
        return retList

    @staticmethod
    def bcacheGetSetUuid(devPath):
        # see C struct definition in makeDevice()
        bcacheSbSetUuidPreFmt = "QQQ16B16B"
        bcacheSbSetUuidFmt = "16B"

        with open(devPath, "rb") as f:
            f.seek(8 * 512 + struct.calcsize(bcacheSbSetUuidPreFmt))
            buf = f.read(struct.calcsize(bcacheSbSetUuidFmt))
            return str(uuid.UUID(bytes=buf))

    @staticmethod
    def scsiGetHostControllerName(devPath):
        devName = os.path.basename(os.path.realpath(devPath))       # XXX -> /dev/sda => sda
        sysfsPath = os.path.join("/sys", "block", devName)          # sda => /sys/block/sda
        hostPath = os.path.realpath(sysfsPath)                      # /sys/block/sda -> /sys/block/devices/pci0000:00/0000:00:17.0/ata3/host2/target2:0:0/2:0:0:0/block/sda
        while True:
            m = re.search("^host[0-9]+$", os.path.basename(hostPath), re.M)
            if m is not None:
                hostControllerNameFile = os.path.join("/sys", "class", "scsi_host", m.group(0), "proc_name")
                return pathlib.Path(hostControllerNameFile).read_text().rstrip()
            hostPath = os.path.dirname(hostPath)
            assert hostPath != "/"

    @staticmethod
    def getBlkDevLvmInfo(devPath):
        """Returns (vg-name, lv-name)
           Returns None if the device is not lvm"""

        ret = subprocess.run(["dmsetup", "info", devPath], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True, text=True)
        if ret.returncode == 0:
            m = re.search("^Name: *(\\S+)$", ret.stdout.rstrip(), re.M)
            assert m is not None
            ret = m.group(1).split(".")
            if len(ret) == 2:
                return ret
            ret = m.group(1).split("-")         # compatible with old lvm version
            if len(ret) == 2:
                return ret

        m = re.fullmatch("(/dev/mapper/\\S+)-(\\S+)", devPath)          # compatible with old lvm version
        if m is not None:
            return Util.getBlkDevLvmInfo("%s-%s" % (m.group(1), m.group(2)))

        return None

    @staticmethod
    def lvmGetSlaveDevPathList(vgName):
        ret = []
        out = Util.cmdCall("/sbin/lvm", "pvdisplay", "-c")
        for m in re.finditer("^\\s*(\\S+):%s:.*" % (vgName), out, re.M):
            if m.group(1) == "[unknown]":
                raise Exception("volume group %s not fully loaded" % (vgName))
            ret.append(m.group(1))
        return ret

    @staticmethod
    def getBlkDevFsType(devPath):
        # FIXME: blkid doesn't support bcachefs yet, use file instead
        ret = Util.cmdCall("file", "-sb", devPath)
        if re.search("^bcachefs, UUID=", ret) is not None:
            return "bcachefs"

        # use blkid to get fstype
        ret = Util.cmdCall("blkid", "-o", "export", devPath)
        m = re.search("^TYPE=(\\S+)$", ret, re.M)
        if m is not None:
            return m.group(1).lower()

        return ""

    @staticmethod
    def devPathIsDiskOrPartition(devPath):
        if re.fullmatch("/dev/sd[a-z]", devPath) is not None:
            return True
        if re.fullmatch("(/dev/sd[a-z])([0-9]+)", devPath) is not None:
            return False
        if re.fullmatch("/dev/xvd[a-z]", devPath) is not None:
            return True
        if re.fullmatch("(/dev/xvd[a-z])([0-9]+)", devPath) is not None:
            return False
        if re.fullmatch("/dev/vd[a-z]", devPath) is not None:
            return True
        if re.fullmatch("(/dev/vd[a-z])([0-9]+)", devPath) is not None:
            return False
        if re.fullmatch("/dev/nvme[0-9]+n[0-9]+", devPath) is not None:
            return True
        if re.fullmatch("(/dev/nvme[0-9]+n[0-9]+)p([0-9]+)", devPath) is not None:
            return False
        assert False

    @staticmethod
    def devPathPartitionToDiskAndPartitionId(partitionDevPath):
        m = re.fullmatch("(/dev/sd[a-z])([0-9]+)", partitionDevPath)
        if m is not None:
            return (m.group(1), int(m.group(2)))
        m = re.fullmatch("(/dev/xvd[a-z])([0-9]+)", partitionDevPath)
        if m is not None:
            return (m.group(1), int(m.group(2)))
        m = re.fullmatch("(/dev/vd[a-z])([0-9]+)", partitionDevPath)
        if m is not None:
            return (m.group(1), int(m.group(2)))
        m = re.fullmatch("(/dev/nvme[0-9]+n[0-9]+)p([0-9]+)", partitionDevPath)
        if m is not None:
            return (m.group(1), int(m.group(2)))
        assert False

    @staticmethod
    def devPathPartitionToDisk(partitionDevPath):
        return Util.devPathPartitionToDiskAndPartitionId(partitionDevPath)[0]

    @staticmethod
    def devPathDiskOrPartitionToDisk(diskOrPartitionDevPath):
        if Util.devPathIsDiskOrPartition(diskOrPartitionDevPath):
            return diskOrPartitionDevPath
        else:
            return Util.devPathPartitionToDiskAndPartitionId(diskOrPartitionDevPath)[0]

    @staticmethod
    def appendOrCreateListFile(path, line):
        if os.path.exists(path):
            buf = pathlib.Path(path).read_text()
            if line not in buf.split("\n"):
                with open(path, "a") as f:
                    if not buf.endswith("\n"):
                        f.write("\n")
                    f.write(line)
                    f.write("\n")
        else:
            with open(path, "w") as f:
                f.write(line)
                f.write("\n")

    @classmethod
    def copyWithSymlink(cls, srcRootDir, dstRootDir, path, uid=None, gid=None, mode=None, exceptionClass=None):
        assert not os.path.isabs(path)
        assert (uid is None and gid is None and mode is None) or (uid is not None and gid is not None and mode is not None)
        if mode is not None:
            assert (mode & 0o0000) == 0

        realSrcRootDir = cls.fullyNormPath(srcRootDir)
        realDstRootDir = cls.fullyNormPath(dstRootDir)
        if exceptionClass is None:
            exceptionClass = Exception

        # uid, gid, mode are not None:
        #   destination file should have metadata as (uid, gid, mode), all parent directories should have metadata as (uid, gid, 0o0755)
        # uid, gid, mode are None:
        #   destination file and all parent directories should have the same metadata as the corresponding source
        while True:
            srcfile = os.path.join(realSrcRootDir, path)
            dstfile = os.path.join(realDstRootDir, path)
            if os.path.normpath(srcfile) != srcfile:
                raise exceptionClass('source "%s" is out of directory "%s"' % (srcfile, srcRootDir))
            if os.path.islink(srcfile):
                cls._copyWithLinkImplLink(realSrcRootDir, realDstRootDir, srcfile, dstfile, path, uid, gid, exceptionClass)
                path = os.path.normpath(os.path.join(os.path.dirname(path), os.readlink(srcfile)))
            elif os.path.isfile(srcfile):
                cls._copyWithLinkImplFile(realSrcRootDir, realDstRootDir, srcfile, dstfile, path, uid, gid, mode, exceptionClass)
                break
            else:
                assert False

    @classmethod
    def _copyWithLinkImplLink(cls, srcRootDir, dstRootDir, srcfile, dstfile, path, uid, gid, exceptionClass):
        if uid is None:
            cls._copyWithLinkImplCopyDirs(dstRootDir, path, srcRootDir, exceptionClass)
            srcSt = os.lstat(srcfile)
            uid, gid = srcSt.st_uid, srcSt.st_gid
        else:
            cls._copyWithLinkImplMakeDirs(dstRootDir, path, srcRootDir, uid, gid, exceptionClass)

        while True:
            if not os.path.islink(dstfile) or os.readlink(dstfile) != os.readlink(srcfile):
                break
            dstSt = os.lstat(dstfile)
            if dstSt.st_uid != uid or dstSt.st_gid != gid:
                os.lchown(dstfile, uid, gid)
            return

        Util.forceDelete(dstfile)
        os.symlink(os.readlink(srcfile), dstfile)
        os.lchown(dstfile, uid, gid)

    @classmethod
    def _copyWithLinkImplFile(cls, srcRootDir, dstRootDir, srcfile, dstfile, path, uid, gid, mode, exceptionClass):
        if uid is None:
            cls._copyWithLinkImplCopyDirs(dstRootDir, path, srcRootDir, exceptionClass)
        else:
            cls._copyWithLinkImplMakeDirs(dstRootDir, path, srcRootDir, uid, gid, exceptionClass)

        while True:
            if not os.path.isfile(dstfile) or os.path.islink(dstfile) or not filecmp.cmp(srcfile, dstfile, shallow=False):
                break
            if uid is None:
                srcSt = os.lstat(srcfile)
                realUid, realGid, realMode = srcSt.st_uid, srcSt.st_gid, (srcSt.st_mode & 0o7777)
            else:
                realUid, realGid, realMode = uid, gid, mode
            dstSt = os.lstat(dstfile)
            if dstSt.st_uid != realUid or dstSt.st_gid != realGid:
                os.lchown(dstfile, realUid, realGid)
            if (dstSt.st_mode & 0o7777) != realMode:
                os.chmod(dstfile, realMode)
            return

        if uid is None:
            shutil.copy2(srcfile, dstfile)
        else:
            shutil.copy(srcfile, dstfile)
            os.lchown(dstfile, uid, gid)
            os.chmod(dstfile, mode)

    @classmethod
    def _copyWithLinkImplCopyDirs(cls, dstRootDir, path, srcRootDir, exceptionClass):
        for p in cls.realPathSplit(path)[:-1]:
            dstRootDir = os.path.join(dstRootDir, p)
            srcRootDir = os.path.join(srcRootDir, p)
            srcSt = os.lstat(srcRootDir)
            if os.path.exists(dstRootDir):
                if os.path.islink(dstRootDir):
                    dstSt = os.lstat(dstRootDir)
                    if dstSt.st_uid != srcSt.st_uid or dstSt.st_gid != srcSt.st_gid:
                        os.lchown(dstRootDir, srcSt.st_uid, srcSt.st_gid)
                elif os.path.isdir(dstRootDir):
                    dstSt = os.lstat(dstRootDir)
                    if dstSt.st_uid != srcSt.st_uid or dstSt.st_gid != srcSt.st_gid:
                        os.lchown(dstRootDir, srcSt.st_uid, srcSt.st_gid)
                    if (dstSt.st_mode & 0o7777) != (srcSt.st_mode & 0o7777):
                        os.chmod(dstRootDir, srcSt.st_mode & 0o7777)
                else:
                    raise exceptionClass('target for "%s" is not a directory' % (srcRootDir))
            else:
                os.mkdir(dstRootDir)
                os.lchown(dstRootDir, srcSt.st_uid, srcSt.st_gid)
                os.chmod(dstRootDir, srcSt.st_mode & 0o7777)

    @classmethod
    def _copyWithLinkImplMakeDirs(cls, dstRootDir, path, srcRootDir, uid, gid, exceptionClass):
        for p in cls.realPathSplit(path)[:-1]:
            dstRootDir = os.path.join(dstRootDir, p)
            srcRootDir = os.path.join(srcRootDir, p)
            if os.path.exists(dstRootDir):
                if os.path.islink(dstRootDir):
                    dstSt = os.lstat(dstRootDir)
                    if dstSt.st_uid != uid or dstSt.st_gid != gid:
                        os.lchown(dstRootDir, uid, gid)
                elif os.path.isdir(dstRootDir):
                    dstSt = os.lstat(dstRootDir)
                    if dstSt.st_uid != uid or dstSt.st_gid != gid:
                        os.lchown(dstRootDir, uid, gid)
                    if (dstSt.st_mode & 0o7777) != 0o0755:
                        os.chmod(dstRootDir, 0o0755)
                else:
                    raise exceptionClass('target for "%s" is not a directory' % (srcRootDir))
            else:
                os.mkdir(dstRootDir)
                os.lchown(dstRootDir, uid, gid)
                os.chmod(dstRootDir, 0o0755)


class TempChdir:

    def __init__(self, dirname):
        self.olddir = os.getcwd()
        os.chdir(dirname)

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        os.chdir(self.olddir)


class PhysicalDiskMounts:

    """This class is a better psutil.disk_partitions()"""

    class Entry:

        def __init__(self, p):
            self._p = p

        @property
        def device(self):
            return self._p.device

        @property
        def mountpoint(self):
            return self._p.mountpoint

        @property
        def fstype(self):
            return self._p.fstype

        @property
        def opts(self):
            return self._p.opts

        @property
        def mnt_opt_list(self):
            return self._p.opts.split(",")

        def __repr__(self):
            return "<%s %r>" % (self.__class__.__name__, self.__dict__)

    class NotFoundError(Exception):
        pass

    @classmethod
    def get_entries(cls):
        return [cls.Entry(p) for p in psutil.disk_partitions()]

    @classmethod
    def find_root_entry(cls):
        ret = cls.find_entry_by_mount_point("/")
        if ret is None:
            raise cls.NotFoundError("no rootfs mount point")
        else:
            return ret

    @classmethod
    def find_entry_by_mount_point(cls, mount_point_path):
        for p in psutil.disk_partitions():
            if p.mountpoint == mount_point_path:
                return cls.Entry(p)
        return None


class RangeMatcher:

    def __init__(self, wildcardList, matchFunc=None):
        self._wildcardList = wildcardList

        if matchFunc is not None:
            self._matchFunc = matchFunc
        else:
            self._matchFunc = self.__defaultMatchFunc

        # validate self._wildcardList
        lastIndex = None
        for i, w in enumerate(self._wildcardList):
            if w == "*":
                continue
            assert i <= 1 or i - lastIndex <= 2                                                 # consecutive "*" found
            assert lastIndex is None or self._matchFunc(self._wildcardList[lastIndex], w) < 0   # versions are not in ascend order
            lastIndex = i

    def match(self, ver):
        lastIsWildcard = False

        for entry in self._wildcardList:
            if entry == "*":
                lastIsWildcard = True
                continue

            ret = self._matchFunc(ver, entry)
            if ret == 0:
                # it is a match if 'ver' exactly matches the current entry in the list
                return True
            elif ret < 0:
                # 'ver' is less than the current entry in the list, and the last entry ('*') allows this,
                # then it matches, or else it doesn't
                return lastIsWildcard
            else:  # ret > 0
                lastIsWildcard = False

        # the loop completes without returning, means 'ver' is greater than the last entry,
        # the last entry is a '*', means self._entryList allows `ver` be greater than the last specified version
        # so it matches, or else it doesn't
        return lastIsWildcard

    def filter(self, verList):
        return [x for x in verList if self.match(x)]

    @staticmethod
    def __defaultMatchFunc(a, b):
        if a == b:
            return 0
        elif a < b:
            return -1
        else:  # a > b
            return 1
